package com.china.produce.message;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.json.JSONUtil;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

/**
 * @ClassName
 * @Author niujie
 * @Version
 * @Description
 * @CreateTime 2025/6/24
 */
@Component
public class RedisSseEmitterMessageListener implements MessageListener {

    private static final String CHANNEL = "sse:message:channel";

    private final StringRedisTemplate redisTemplate;

    public RedisSseEmitterMessageListener(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = redisTemplate.getStringSerializer().deserialize(message.getBody());
        MessageDO msg = JSONUtil.toBean(body, MessageDO.class);
        if (Boolean.TRUE.equals(msg.getNeedBroadcast())) {
            SseEmitterMessageSender.broadcast(msg.getMessageText());
        } else if (CollUtil.isNotEmpty(msg.getSessionKeys())) {
            msg.getSessionKeys().forEach(key -> SseEmitterMessageSender.send(key, msg.getMessageText()));
        }
    }

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory factory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener(this, new ChannelTopic(CHANNEL));
        return container;
    }
}
